package cn.xiaochao.mapper;

import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.HashMap;
import java.util.Map;

import cn.xiaochao.zebra.common.OwnEnv;
import cn.xiaochao.zebra.common.ZKConnectRunner;
import rpc.domain.FileSplit;
import rpc.domain.HttpAppHost;

public class MapperRunner implements Runnable {

	@Override
	public void run() {
		// TODO Auto-generated method stub
		while (true) {
			//保存的是处理后的信息
			Map<CharSequence,HttpAppHost> map = new HashMap<CharSequence,HttpAppHost>();
			// 1.取出切片
			FileInputStream in = null;
			FileChannel fileChannel = null;
			try {
				System.out.println("一级引擎01开始处理切片");
				// 要从一级引擎自己的切片队列中获取一个切片
				FileSplit split = OwnEnv.getSpiltQueue().take();
				//将当前的一级引擎改为busy
				ZKConnectRunner.setBusy();
				// 2.切片时存在不是从行首或者行尾切得,所以需要追溯
				// start与end 都采用向前追溯

				File logFile = new File(split.getPath().toString());

				long start = split.getStart();
				long end = start + split.getLength();
				// 3.追溯
			    in = new FileInputStream(logFile);

				fileChannel = in.getChannel();

				if (start != 0) {

					long headerPostion = start;
					while (true) {

						// 读取一个字节
						ByteBuffer buf = ByteBuffer.allocate(1);
						fileChannel.position(headerPostion);
						fileChannel.read(buf);
						// 向前追溯到上一个/n换行节点
						if (new String(buf.array()).equals("\n")) {
							
							start = headerPostion + 1;
							break;
							
						} else {
							
							headerPostion--;
							
						}

					}

				}
				if (end != logFile.length()) {

					long endPostion = end;
					
					while (true) {

						// 读取一个字节
						ByteBuffer buf = ByteBuffer.allocate(1);
						fileChannel.position(endPostion);
						fileChannel.read(buf);
						// 向前追溯到上一个/n换行节点
						if (new String(buf.array()).equals("\n")) {
							
							end = endPostion;
							break;
							
						} else {
							
							endPostion--;
							
						}
				
					}
					
				}
				//设置起始位置 (造成死循环)
								
				fileChannel.position(start);
				
				ByteBuffer dataBackBuf = ByteBuffer.allocate((int)(end - start));
				fileChannel.read(dataBackBuf);
				//byteBuffer 读取为BufferReader
				BufferedReader reader = new BufferedReader(
						new InputStreamReader(
								new ByteArrayInputStream(dataBackBuf.array())));
				String line = null;
				
				
				while((line=reader.readLine())!=null){
					
					//读取到一个对象
					String[] data = line.split("\\|");
					HttpAppHost hah = dataSwitchHttpAppHost(logFile, data);
					//声明代表一个日志唯一的key
					CharSequence key=hah.getReportTime() + "|" + hah.getAppType() + "|" + hah.getAppSubtype() + "|" + hah.getUserIP() + "|" + hah.getUserPort() + "|" + hah.getAppServerIP() + "|" + hah.getAppServerPort() +"|" + hah.getHost() + "|" + hah.getCellid();
					
					if(map.containsKey(key)){
						HttpAppHost mapHah=map.get(key);
						mapHah.setAccepts(mapHah.getAccepts()+hah.getAccepts());
						mapHah.setAttempts(mapHah.getAttempts()+hah.getAttempts());
						mapHah.setTrafficUL(mapHah.getTrafficUL()+hah.getTrafficUL());
						mapHah.setTrafficDL(mapHah.getTrafficDL()+hah.getTrafficDL());
						mapHah.setRetranUL(mapHah.getRetranUL()+hah.getRetranUL());
						mapHah.setRetranDL(mapHah.getRetranDL()+hah.getRetranDL());
						mapHah.setTransDelay(mapHah.getTransDelay()+hah.getTransDelay());
						map.put(key, mapHah);
					}else{
						map.put(key,hah);
					}
					
				}
				
				System.out.println("切块处理后："+map.size());
				OwnEnv.getMapQueue().add(map);
				//将状态该为free
				ZKConnectRunner.setFree();
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (FileNotFoundException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} finally {
				
				try {
					in.close();
					fileChannel.close();
					
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				
			}

		}

	}

	private HttpAppHost dataSwitchHttpAppHost(File logFile, String[] data) {
		HttpAppHost hah=new HttpAppHost();
		String reportTime=logFile.getName().split("_")[1];
		hah.setReportTime(reportTime);
		//上网小区的id
		hah.setCellid(data[16]);
		//应用大类
		hah.setAppType(Integer.parseInt(data[22]));
		//应用子类
		hah.setAppSubtype(Integer.parseInt(data[23]));
		//用户ip
		hah.setUserIP(data[26]);
		//用户port
		hah.setUserPort(Integer.parseInt(data[28]));
		//访问的服务ip
		hah.setAppServerIP(data[30]);
		//访问的服务port
		hah.setAppServerPort(Integer.parseInt(data[32]));
		//域名
		hah.setHost(data[58]);
		//业务类型编码
		int appTypeCode=Integer.parseInt(data[18]);
		String transStatus=data[54];
		//--将需要计算的属性，按照业务规则通过辅助信息进行计算后设置到hah中
		//----业务规则1:如果没有小区编号，就将小区编号补齐为000000000
		if(hah.getCellid()==null||hah.getCellid().equals("")){
			hah.setCellid("000000000");
		}
		//----业务规则2:计算尝试次数 - attempts if( App Type Code=103 ) 1
		if(appTypeCode==103){
			hah.setAttempts(1);
		}
		//----业务规则3:接受次数 - accepts if( App Type Code=103 & HTTP/WAP事物状态 in(10,11,...,200,201,202,203,204,205,206,302,304,306) && Wtp中断类型==NULL) then counter++ 1
		if(appTypeCode==103 &&"10,11,12,13,14,15,32,33,34,35,36,37,38,48,49,50,51,52,53,54,55,199,200,201,202,203,204,205 ,206,302,304,306".contains(transStatus)
				&&data[67]==null){
			hah.setAccepts(1);
		}else{
			hah.setAccepts(0);
		}
		if(appTypeCode == 103){
			hah.setTrafficUL(Long.parseLong(data[33]));
			hah.setTrafficDL(Long.parseLong(data[34]));
			hah.setRetranUL(Long.parseLong(data[39]));
			hah.setRetranDL(Long.parseLong(data[40]));
			//业务规则4:传输延时=请求结束-请求起始
			hah.setTransDelay(Long.parseLong(data[20]) -Long.parseLong(data[19]));
		}
		return hah;
	}

}
